package com.flink;

import com.flink.vo.BankVo;
import com.flink.vo.DiffType;
import com.flink.vo.MergeVo;
import com.flink.vo.PayOrgVo;
import org.apache.flink.api.common.functions.CoGroupFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.CoGroupOperator;
import org.apache.flink.util.Collector;

/**
 * 利用coGroup实现对账需求<br/>
 */
public class BatchJob3 {
    public static void main(String[] args) throws Exception {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        DataSet<PayOrgVo> payOrgSource = env.fromElements(new PayOrgVo("113", 1), new PayOrgVo("000", 2), new PayOrgVo("115", 33));
        DataSet<BankVo> bankSource = env.fromElements(new BankVo("000", 2), new BankVo("115", 333), new BankVo("114", 4));

        CoGroupOperator<PayOrgVo, BankVo, MergeVo> merge = payOrgSource.coGroup(bankSource)
                .where(PayOrgVo::getOrderNo)//利用关联字段把两边的数据关联起来
                .equalTo(BankVo::getOrderNo)
                .with(new CoGroupFunction<PayOrgVo, BankVo, MergeVo>() {// with方法会将两边orderNo相同的数据放在同一个方法中处理
                    @Override
                    public void coGroup(Iterable<PayOrgVo> first, Iterable<BankVo> second, Collector<MergeVo> out) throws Exception {
                        // 进入到coGroup方法的数据都是orderNo相同的，如果关联不上的即为空
                        PayOrgVo payOrgVo = null;
                        BankVo bankVo = null;
                        DiffType diffType = null;
                        // 以下代码假定数据具有唯一性，即同一个orderNo下仅有一条数据
                        // 如果重复数据，根据实际情况写下面的代码逻辑
                        for (PayOrgVo vo : first) {
                            payOrgVo = vo;
                        }
                        for (BankVo vo : second) {
                            bankVo = vo;
                        }

                        if (bankVo == null) {// 相同orderNo下，支付机构有数据
                            diffType = DiffType.F113;
                        } else if (payOrgVo == null) {// 银行有数据
                            diffType = DiffType.F114;
                        } else if (payOrgVo.getPayment().equals(bankVo.getPayment())) {// 数据完全一致
                            diffType = DiffType.F000;
                        } else {// orderNo相同但payment不同
                            diffType = DiffType.F115;
                        }
                        // 返回数据
                        out.collect(new MergeVo(diffType, payOrgVo, bankVo));
                    }
                });
        merge.print();
    }
}
